package realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import realtime.common.EduConfig;
import realtime.util.HbaseUtil;

/**
 * richsinkfunction,可以通过该方法中的invoke将流数据写入hbase
 * @Create 2023/9/5 21:16
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn=HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        String type = value.getString("type");
        JSONObject tableProcess = value.getJSONObject("table_process");

        if (type.equals("delete")){
            //System.out.println("删除数据");
            HbaseUtil.delRow(conn, EduConfig.HBASE_NAMESPACE,tableProcess.getString("sink_table"),value.getString(tableProcess.getString("sink_pk")));
        }else {
            //System.out.println("插入或更新数据");
            String[] columns = tableProcess.getString("sink_columns").trim().split(",");
            String[] values = new String[columns.length];

            for (int i = 0; i < columns.length; i++) {
                values[i]=value.getString(columns[i]);
            }
            HbaseUtil.putRow(conn,EduConfig.HBASE_NAMESPACE,tableProcess.getString("sink_table"),value.getString(tableProcess.getString("sink_pk")),
                    "info",columns,values);

        }

        //如果对维度表做了删除或者修改操作，将可能存在于Redis中的数据删除掉
    }
}
